#pragma once
#include <asio/io_context.hpp>
#include <asio/use_awaitable.hpp>
#include <asio/detached.hpp>
#include <asio/awaitable.hpp>
#include <asio/co_spawn.hpp>
#include <grpc++/grpc++.h>
#include <thread>
#include <tuple>
#include "proto_server_context.hpp"
template<typename MemFunc>
struct async_func_helper;

// std::unique_ptr< ::grpc::ClientAsyncWriterInterface< ::int_message>> Asyncstream_param(::grpc::ClientContext* context, ::int_message* response, ::grpc::CompletionQueue* cq, void* tag)
template<typename ReqType,typename RespType,typename StubClass>
struct async_func_helper<
std::unique_ptr<grpc::ClientAsyncResponseReader<RespType>>(
            StubClass::*)(grpc::ClientContext*,const ReqType&,grpc::CompletionQueue*)>
{
    using func_type = std::function<asio::awaitable<RespType>(const ReqType&)>;
};
// stream param normal return
template<typename ReqType,typename RespType,typename StubClass>
struct async_func_helper<
std::unique_ptr<grpc::ClientAsyncWriter<ReqType>>(
            StubClass::*)(grpc::ClientContext*,RespType*,grpc::CompletionQueue*,void*)>
{
    using TupleType = std::tuple<std::function<asio::awaitable<void>(std::optional<ReqType>)>,std::function<asio::awaitable<RespType>()>>;
    using func_type = std::function<asio::awaitable<TupleType>()>;
};
// normal param stream return
template<typename ReqType,typename RespType,typename StubClass>
struct async_func_helper<
std::unique_ptr<grpc::ClientAsyncReader<RespType>>(
            StubClass::*)(grpc::ClientContext*,const ReqType&,grpc::CompletionQueue*,void*)>
{
    using return_type = asio::awaitable<std::function<asio::awaitable<std::optional<RespType>>()>>;
    using func_type = std::function<return_type(const ReqType&)>;
};
// stream param stream return
template<typename ReqType,typename RespType,typename StubClass>
struct async_func_helper<
std::unique_ptr<grpc::ClientAsyncReaderWriter<ReqType,RespType>>(
            StubClass::*)(grpc::ClientContext*,grpc::CompletionQueue*,void*)>
{
    using sender_type = std::function<asio::awaitable<void>(std::optional<ReqType>)>;
    using reciver_type = std::function<asio::awaitable<std::optional<RespType>>()>;
    using tuple_type = std::tuple<
        sender_type,
        reciver_type
    >;
    using return_type = asio::awaitable<tuple_type>;
    using func_type = std::function<return_type()>;
};

template<typename ParamType>
struct stream_param_init;

template<>
struct stream_param_init<void>{
    template<typename Handler>
    void operator()(Handler&& h,std::function<void(std::function<void()>)> wrap_func)
    {
        struct handler_ctx{
            Handler h;
            bool executed = false;
        };
        auto _ctx = std::make_shared<handler_ctx>(std::move(h));
        wrap_func(std::bind([](decltype(_ctx) ctx){
            if(ctx->executed){
                throw std::runtime_error("error duplicated excution of coroutine handler");
            }
            ctx->h();
            ctx->executed = true;
        },_ctx));
    }
};
template<typename ParamType>
struct stream_param_init{
    template<typename Handler>
    void operator()(Handler&& h,std::function<void(std::function<void(ParamType)>)> wrap_func)
    {
        struct handler_ctx{
            Handler h;
        };
        auto _ctx = std::make_shared<handler_ctx>(std::move(h));
        wrap_func(std::bind([](ParamType param,decltype(_ctx) ctx){
            ctx->h(std::move(param));
        },std::placeholders::_1,_ctx));
    }
};
class GrpcMetaDataService:public asio::io_context::service{
private:
    server_context _client_metadata;
public:
    GrpcMetaDataService(asio::io_context& _ctx):asio::io_context::service(_ctx){}

    server_context* get_metadata(){
        return &_client_metadata;
    }

    static inline asio::io_context::id id;
};
class GrpcCompletionqueueService:public asio::io_context::service{
private:
    grpc::CompletionQueue _queue;
    unsigned long tag_seed = 1;
    std::thread working_thread;
    asio::io_context& _ctx = get_io_context();
    decltype(asio::make_work_guard(_ctx)) work_guark = asio::make_work_guard(_ctx);
    std::map<void*,std::function<void(bool)>> _response_map;

public:
    GrpcCompletionqueueService(asio::io_context& ctx):
        asio::io_context::service(ctx){

        working_thread = std::thread([this](){
            bool ok = true;
            bool running = true;
            void* got_tag;
            while(running)
            {
                running = _queue.Next(&got_tag,&ok);
                asio::post(_ctx,[this,got_tag,ok](){
                    // std::cout<<"new event "<<got_tag<<std::endl;
                    if(auto ite = _response_map.find(got_tag);ite != _response_map.end())
                    {
                        ite->second(ok);
                        _response_map.erase(ite);
                    }
                });
            }
            std::cout<<"client exited"<<std::endl;
        });
    }
    static inline asio::io_context::id id;


    void shutdown() final{
        std::cout<<"queue shudown"<<std::endl;
        work_guark.reset();
        _queue.Shutdown();
        working_thread.join();
        _response_map.clear();
    }
    decltype(_response_map)& get_map(){
        return _response_map;
    }

    grpc::CompletionQueue& get_completion_queue(){
        return _queue;
    }
    void* generate_tag(){
        return (void*)(tag_seed++);
    }
};
template<typename ClientClass>
class GrpcClientService:public asio::io_context::service{
protected:
    using super = GrpcClientService<ClientClass>;
    using super_consctrunctor = GrpcClientService<ClientClass>::GrpcClientService;
    using target_class = ClientClass;
private:
    asio::io_context& ctx = get_io_context();
    decltype(asio::make_work_guard(ctx)) work_guark = asio::make_work_guard(ctx);
    GrpcCompletionqueueService& _queue_service = asio::use_service<GrpcCompletionqueueService>(ctx);
    GrpcMetaDataService& _meta_data_service = asio::use_service<GrpcMetaDataService>(ctx);
    grpc::CompletionQueue& _completion_queue = _queue_service.get_completion_queue();
    std::map<void*,std::function<void(bool)>>& _response_map = _queue_service.get_map();
    std::unique_ptr<typename ClientClass::Stub> _stub;

    void fill_meta_data(grpc::ClientContext& context){
        for(const auto key: *_meta_data_service.get_metadata())
        {
            context.AddMetadata(key.first,key.second);
        }
    }

public:
    GrpcClientService(asio::io_context& ctx):
        asio::io_context::service(ctx){

    }
    virtual ~GrpcClientService(){
    }

    void* generate_tag(){
        return _queue_service.generate_tag();
    }
    void connect(std::shared_ptr<grpc::Channel> channel)
    {
        if(_stub != nullptr)
        {
            throw std::logic_error("duplicated connected!");
        }
        _stub = ClientClass::NewStub(channel);
    }
    void disconnect()
    {
        if(_stub == nullptr)
        {
            throw std::logic_error("error connection is invalid, cannot disconnect");
        }
        _stub.reset();
    }

    void shutdown() final{
        std::cout<<"service shudown"<<std::endl;
        _stub.reset();
        work_guark.reset();
    }

    // normal param normal return
    template<typename ReqType,typename RespType>
    auto
        bind_client(std::unique_ptr<grpc::ClientAsyncResponseReader<RespType>>(
            ClientClass::Stub::* async_func)(grpc::ClientContext*,const ReqType&,grpc::CompletionQueue*)
        )->typename async_func_helper<decltype(async_func)>::func_type
        {
            return [this,async_func](const ReqType& req)->asio::awaitable<RespType>{
                grpc::Status status;
                grpc::ClientContext client_context;
                fill_meta_data(client_context);
                RespType response;
                auto rpc_call = (_stub.get()->*async_func)(&client_context,req,&_completion_queue);
                co_await asio::async_initiate<
                    decltype(asio::use_awaitable),
                    void()>(
                        stream_param_init<void>{},asio::use_awaitable,
                        [this,&rpc_call,&response,&status](std::function<void()> callback){
                            void* key = generate_tag();
                            _response_map[key]=[callback](bool){
                                callback();
                            };
                            rpc_call->Finish(&response,&status,key);
                        }
                );
                
                if(!status.ok())
                {
                    throw status;
                }
                co_return std::move(response);
            };
        }

    
    // stream param normal return
    template<typename ReqType,typename RespType>
    auto
        bind_client(std::unique_ptr<grpc::ClientAsyncWriter<ReqType>>(
            ClientClass::Stub::* async_func)(grpc::ClientContext*,RespType*,grpc::CompletionQueue*,void*)
        )->typename async_func_helper<decltype(async_func)>::func_type
        {
            using TupleType = typename async_func_helper<decltype(async_func)>::TupleType;
            struct stream_param_context{
                std::unique_ptr<grpc::ClientAsyncWriter<ReqType>> writer;
                grpc::ClientContext context;
                RespType response;
            };
            return [this,async_func]()->asio::awaitable<TupleType>{
                auto _ctx = std::make_shared<stream_param_context>();
                fill_meta_data(_ctx->context);
                auto init_success = co_await asio::async_initiate<
                            decltype(asio::use_awaitable),
                            void(bool)>(
                                stream_param_init<bool>{},asio::use_awaitable,
                                [this,_ctx,async_func](std::function<void(bool)> callback){
                                    auto start_tag = generate_tag();
                                    _response_map[start_tag]=[callback,_ctx](bool ok){
                                        callback(ok);
                                    };
                                    _ctx->writer = (_stub.get()->*async_func)(&_ctx->context,&_ctx->response,&_completion_queue,start_tag);
                                });
                if(!init_success){
                    throw grpc::Status(grpc::StatusCode::UNAVAILABLE,"start rpc call failed");
                }
                co_return TupleType{
                    [_ctx,this](std::optional<ReqType> data)->asio::awaitable<void>{
                        auto success = co_await asio::async_initiate<
                            decltype(asio::use_awaitable),
                            void(bool)>(
                                stream_param_init<bool>{},asio::use_awaitable,
                                [&_ctx,this,data](std::function<void(bool)> callback){
                                        auto send_tag = generate_tag();
                                        _response_map[send_tag]=[callback](bool status){
                                            callback(status);
                                        };
                                        if(data){
                                            _ctx->writer->Write(*data,send_tag);
                                        }else{
                                            _ctx->writer->WritesDone(send_tag);
                                        }
                                });
                        if(!success){
                            auto s = co_await async_wrapper<grpc::Status>(
                                [this,&_ctx](std::function<void(grpc::Status)> callback){
                                    auto finish_tag = generate_tag();
                                    auto Res_ptr = std::make_shared<grpc::Status>();
                                    _response_map[finish_tag]=[callback,Res_ptr,_ctx](bool ok){
                                        callback(std::move(*Res_ptr));
                                    }; 
                                    _ctx->writer->Finish(Res_ptr.get(),finish_tag);
                                }
                            );
                            if(!s.ok()){
                                throw s;
                            }else{
                                throw grpc::Status(grpc::StatusCode::UNAVAILABLE,"send data failed");
                            }
                        }
                    },
                    [_ctx,this]()->asio::awaitable<RespType>{
                        grpc::Status s;
                        co_await asio::async_initiate<
                            decltype(asio::use_awaitable),
                            void()>(
                                stream_param_init<void>{},asio::use_awaitable,
                                [&_ctx,&s,this](std::function<void()> callback){
                                    auto finish_tag = generate_tag();
                                    _response_map[finish_tag]=[_ctx,callback](bool){
                                        callback();
                                    };

                                    _ctx->writer->Finish(&s,finish_tag);
                                });
                        if(!s.ok()){
                            throw s;
                        }
                        co_return std::move(_ctx->response);
                    }
                };
            };
        }

        template<typename ReturnType>
        asio::awaitable<ReturnType> async_wrapper(std::function<void(std::function<void(ReturnType)>)> process_func){
            return asio::async_initiate<
                                decltype(asio::use_awaitable),
                                void(ReturnType)>(
                                    stream_param_init<ReturnType>{},asio::use_awaitable,
                                    std::move(process_func));
        }
    // normal param stream return
    template<typename ReqType,typename RespType>
    auto
        bind_client(std::unique_ptr<grpc::ClientAsyncReader<RespType>>(
            ClientClass::Stub::* async_func)(grpc::ClientContext*,const ReqType&,grpc::CompletionQueue*,void*)
        )->typename async_func_helper<decltype(async_func)>::func_type
        {            
            struct stream_param_context{
                std::unique_ptr<grpc::ClientAsyncReader<RespType>> reader;
                grpc::ClientContext context;
            };
            return [this,async_func](const ReqType& req)->typename async_func_helper<decltype(async_func)>::return_type{
                
                auto _ctx = std::make_shared<stream_param_context>();
                fill_meta_data(_ctx->context);
                auto init_success = co_await async_wrapper<bool>(
                                [this,_ctx,async_func,&req](std::function<void(bool)> callback){
                                    auto start_tag = generate_tag();
                                    _response_map[start_tag]=[callback,_ctx](bool ok){
                                        callback(ok);
                                    };
                                    _ctx->reader = (_stub.get()->*async_func)(&_ctx->context,req,&_completion_queue,start_tag);
                                });
                if(!init_success){
                    throw grpc::Status(grpc::StatusCode::UNAVAILABLE,"start rpc call failed");
                }

                co_return [this,_ctx]()->asio::awaitable<std::optional<RespType>>{
                    auto response = co_await async_wrapper<std::optional<RespType>>(
                                    [this,&_ctx](std::function<void(std::optional<RespType>)> callback){
                                        auto read_tag = generate_tag();
                                        auto Res_ptr = std::make_shared<RespType>();
                                        _response_map[read_tag]=[callback,_ctx,Res_ptr](bool ok){
                                            callback(ok?*Res_ptr:std::optional<RespType>{});
                                        };
                                        _ctx->reader->Read(Res_ptr.get(),read_tag);
                                    });
                    if(!response){
                        auto s = co_await async_wrapper<grpc::Status>(
                            [this,&_ctx](std::function<void(grpc::Status)> callback){
                                auto finish_tag = generate_tag();
                                auto Res_ptr = std::make_shared<grpc::Status>();
                                _response_map[finish_tag]=[callback,Res_ptr,_ctx](bool ok){
                                    callback(std::move(*Res_ptr));
                                }; 
                                _ctx->reader->Finish(Res_ptr.get(),finish_tag);
                            }
                        );

                        if(!s.ok()){
                            throw s;
                        }
                    }
                    co_return std::move(response);
                };
            };
        }
    // stream param stream return
    template<typename ReqType,typename RespType>
    auto
        bind_client(std::unique_ptr<grpc::ClientAsyncReaderWriter<ReqType,RespType>>(
            ClientClass::Stub::* async_func)(grpc::ClientContext*,grpc::CompletionQueue*,void*)
        )->typename async_func_helper<decltype(async_func)>::func_type
        {            
            struct stream_param_context{
                std::unique_ptr<grpc::ClientAsyncReaderWriter<ReqType,RespType>> rw;
                grpc::ClientContext context;
            };
            using func_helper = async_func_helper<decltype(async_func)>;
            using tuple_type = typename async_func_helper<decltype(async_func)>::tuple_type;
            using return_type = typename async_func_helper<decltype(async_func)>::return_type;
            return [this,async_func]()->return_type{
                auto _ctx = std::make_shared<stream_param_context>();
                fill_meta_data(_ctx->context);
                auto init_success = co_await async_wrapper<bool>(
                                [this,&_ctx,async_func](std::function<void(bool)> callback){
                                    auto start_tag = generate_tag();
                                    _response_map[start_tag]=[callback](bool ok){
                                        callback(ok);
                                    };
                                    _ctx->rw = (_stub.get()->*async_func)(&_ctx->context,&_completion_queue,start_tag);
                                });
                if(!init_success){
                    throw grpc::Status(grpc::StatusCode::UNAVAILABLE,"start rpc call failed");
                }
                co_return tuple_type{
                    [_ctx,this](std::optional<ReqType> data)->asio::awaitable<void>{
                        auto result = co_await async_wrapper<bool>(
                                [this,&_ctx,&data](std::function<void(bool)> callback){
                                    auto send_tag = generate_tag();
                                    _response_map[send_tag]=[callback](bool ok){
                                        callback(ok);
                                    };
                                    if(data){
                                        _ctx->rw->Write(*data,send_tag);
                                    }else{
                                        _ctx->rw->WritesDone(send_tag);
                                    }
                                });
                        if(!result){
                            auto s = co_await async_wrapper<grpc::Status>(
                                [this,&_ctx](std::function<void(grpc::Status)> callback){
                                    auto finish_tag = generate_tag();
                                    auto Res_ptr = std::make_shared<grpc::Status>();
                                    _response_map[finish_tag]=[callback,Res_ptr,_ctx](bool ok){
                                        callback(std::move(*Res_ptr));
                                    }; 
                                    _ctx->rw->Finish(Res_ptr.get(),finish_tag);
                                }
                            );
                            if(!s.ok()){
                                throw s;
                            }else{
                                throw grpc::Status(grpc::StatusCode::UNAVAILABLE,"send data failed");
                            }
                        }
                    },
                    [this,_ctx]()->asio::awaitable<std::optional<RespType>>{
                        auto response = co_await async_wrapper<std::optional<RespType>>(
                                [this,&_ctx](std::function<void(std::optional<RespType>)> callback){
                                    auto read_tag = generate_tag();
                                    auto data = std::make_shared<RespType>();
                                    _response_map[read_tag]=[data,callback](bool ok){
                                        callback(ok?*data:std::optional<RespType>{});
                                    };
                                    _ctx->rw->Read(data.get(),read_tag);
                                });
                        if(!response){
                            auto s = co_await async_wrapper<grpc::Status>(
                                [this,&_ctx](std::function<void(grpc::Status)> callback){
                                    auto finish_tag = generate_tag();
                                    auto Res_ptr = std::make_shared<grpc::Status>();
                                    _response_map[finish_tag]=[callback,Res_ptr,_ctx](bool ok){
                                        callback(std::move(*Res_ptr));
                                    }; 
                                    _ctx->rw->Finish(Res_ptr.get(),finish_tag);
                                }
                            );
                            if(!s.ok()){
                                throw s;
                            }
                        }                        
                        co_return response;
                    }
                };
            };
        }

};
#define DEF_SERVICE_CONSCTRUCTOR() using super::GrpcClientService
#define BIND_FUNC(func_name) const typename async_func_helper<decltype(&target_class::Stub::Async##func_name)>::func_type func_name = bind_client(&target_class::Stub::Async##func_name)

#define BEGIN_SERVICE(name,base_class) class name : public GrpcClientService<base_class>{\
public: DEF_SERVICE_CONSCTRUCTOR();\
static inline asio::io_context::id id;

#define END_SERVICE(name) };
